package com.reactim.message.connection;

import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * RSocket请求器持有者
 * 管理RSocket连接的本地缓存，由于RSocketRequester不可序列化，需要在本地存储
 */
@Component
@Slf4j
public class RSocketRequesterHolder {
    
    // 存储RSocketRequester的本地缓存 (requesterKey -> RSocketRequester)
    private final ConcurrentHashMap<String, RSocketRequester> requesterCache = new ConcurrentHashMap<>();
    
    // 用户ID到请求器键的映射 (userId -> requesterKey)
    private final ConcurrentHashMap<Long, String> userRequesterMapping = new ConcurrentHashMap<>();
    
    /**
     * 存储RSocketRequester并返回引用键
     * 
     * @param userId 用户ID
     * @param requester RSocket请求器
     * @return 引用键
     */
    public String storeRequester(Long userId, RSocketRequester requester) {
        // 生成唯一的引用键
        String requesterKey = generateRequesterKey(userId);
        
        // 如果用户已有连接，先清理旧连接
        String existingKey = userRequesterMapping.get(userId);
        if (existingKey != null) {
            removeRequester(existingKey);
        }
        
        // 存储新连接
        requesterCache.put(requesterKey, requester);
        userRequesterMapping.put(userId, requesterKey);
        
        log.debug("RSocketRequester已存储: userId={}, requesterKey={}", userId, requesterKey);
        return requesterKey;
    }
    
    /**
     * 根据引用键获取RSocketRequester
     * 
     * @param requesterKey 引用键
     * @return RSocket请求器，如果不存在则返回null
     */
    public RSocketRequester getRequester(String requesterKey) {
        RSocketRequester requester = requesterCache.get(requesterKey);
        
        if (requester != null) {
            // 检查连接是否仍然有效
            if (isRequesterValid(requester)) {
                return requester;
            } else {
                // 连接已失效，移除缓存
                removeRequester(requesterKey);
                log.warn("RSocketRequester已失效，已从缓存中移除: requesterKey={}", requesterKey);
                return null;
            }
        }
        
        return null;
    }
    
    /**
     * 根据用户ID获取RSocketRequester
     * 
     * @param userId 用户ID
     * @return RSocket请求器，如果不存在则返回null
     */
    public RSocketRequester getRequesterByUserId(Long userId) {
        String requesterKey = userRequesterMapping.get(userId);
        if (requesterKey != null) {
            return getRequester(requesterKey);
        }
        return null;
    }
    
    /**
     * 移除RSocketRequester
     * 
     * @param requesterKey 引用键
     */
    public void removeRequester(String requesterKey) {
        RSocketRequester requester = requesterCache.remove(requesterKey);
        
        if (requester != null) {
            // 从用户映射中移除
            userRequesterMapping.entrySet().removeIf(entry -> 
                    requesterKey.equals(entry.getValue()));
            
            // 尝试关闭连接
            try {
                if (!requester.rsocketClient().isDisposed()) {
                    requester.rsocketClient().dispose();
                }
            } catch (Exception e) {
                log.warn("关闭RSocketRequester时发生错误: requesterKey={}, error={}", 
                        requesterKey, e.getMessage());
            }
            
            log.debug("RSocketRequester已移除: requesterKey={}", requesterKey);
        }
    }
    
    /**
     * 根据用户ID移除RSocketRequester
     * 
     * @param userId 用户ID
     */
    public void removeRequesterByUserId(Long userId) {
        String requesterKey = userRequesterMapping.remove(userId);
        if (requesterKey != null) {
            removeRequester(requesterKey);
        }
    }
    
    /**
     * 检查RSocketRequester是否有效
     * 
     * @param requester RSocket请求器
     * @return 是否有效
     */
    private boolean isRequesterValid(RSocketRequester requester) {
        try {
            return requester != null && 
                   requester.rsocketClient() != null && 
                   !requester.rsocketClient().isDisposed();
        } catch (Exception e) {
            log.debug("检查RSocketRequester有效性时发生错误: {}", e.getMessage());
            return false;
        }
    }
    
    /**
     * 生成请求器引用键
     * 
     * @param userId 用户ID
     * @return 引用键
     */
    private String generateRequesterKey(Long userId) {
        return String.format("req_%d_%s_%d", 
                userId, 
                UUID.randomUUID().toString().replace("-", "").substring(0, 8),
                System.currentTimeMillis());
    }
    
    /**
     * 获取当前缓存的连接数量
     * 
     * @return 连接数量
     */
    public int getCachedConnectionCount() {
        return requesterCache.size();
    }
    
    /**
     * 清理所有无效的连接
     */
    public void cleanupInvalidConnections() {
        int initialSize = requesterCache.size();
        
        requesterCache.entrySet().removeIf(entry -> {
            String requesterKey = entry.getKey();
            RSocketRequester requester = entry.getValue();
            
            if (!isRequesterValid(requester)) {
                // 从用户映射中移除
                userRequesterMapping.entrySet().removeIf(userEntry -> 
                        requesterKey.equals(userEntry.getValue()));
                
                log.debug("清理无效连接: requesterKey={}", requesterKey);
                return true;
            }
            return false;
        });
        
        int cleanedCount = initialSize - requesterCache.size();
        if (cleanedCount > 0) {
            log.info("清理无效连接完成，清理数量: {}", cleanedCount);
        }
    }
    
    /**
     * 获取所有在线用户ID
     * 
     * @return 用户ID集合
     */
    public java.util.Set<Long> getOnlineUserIds() {
        return new java.util.HashSet<>(userRequesterMapping.keySet());
    }
}